package com.zhang.hadoop.flink.test11;

import com.zhang.hadoop.flink.test10.LoginEvent;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author: zhang yufei
 * @createTime:2022/12/10 15:55
 * @description:
 */
public class NFAExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.获取登录数据流
        KeyedStream<LoginEvent, String> loginEventStream = env.fromElements(
                new LoginEvent("ouyanghui", "yindao", "fail", 2000L),
                new LoginEvent("ouyanghui", "gangmen", "fail", 3000L),
                new LoginEvent("ouyanghui", "gangmen", "fail", 4000L),
                new LoginEvent("chenyuping", "zujiao", "fail", 5000L),
                new LoginEvent("chenyuping", "koujiao", "fail", 8000L),
                new LoginEvent("chenyuping", "gangmen", "fail", 100000L),
                new LoginEvent("yangdan", "yindao", "fail", 2000L),
                new LoginEvent("yangdan", "gangmen", "fail", 3000L),
                new LoginEvent("yangdan", "gangmen", "success", 4000L)
        ).keyBy(event -> event.userId);

        //2.数据按照顺序一次输入，用户状态机进行处理，状态跳转
        SingleOutputStreamOperator<String> warningStream = loginEventStream.flatMap(new StateMachineMapper());

        warningStream.print();

        env.execute();
    }

    //实现自定的RichFlatMapFunction
    public static class StateMachineMapper extends RichFlatMapFunction<LoginEvent, String> {

        //声明状态机当前的状态
        ValueState<State> currentState;

        @Override
        public void open(Configuration parameters) throws Exception {
            currentState = getRuntimeContext().getState(new ValueStateDescriptor<State>("satte", State.class));
        }

        @Override
        public void flatMap(LoginEvent loginEvent, Collector<String> collector) throws Exception {
            //如果状态为空，进行初始化
            State state = currentState.value();
            if (state == null) {
                state = State.initial;
            }

            //跳转到下一状态
            State nextState = state.transition(loginEvent.eventType);

            //判断当前状态的特殊情况，直接进行跳转
            if (nextState == State.Matched) {
                //检测到了匹配，输出报警信息：不更新状态就是跳转回S2
                collector.collect(loginEvent.userId + "连续三次登录失败");
            } else if (nextState == State.Terminal) {
                //直接将状态更新为初始状态，重新开始检测
                currentState.update(State.initial);
            } else {
                //状态覆盖跳转
                currentState.update(nextState);
            }
        }
    }

    //状态机实现
    public enum State {
        Terminal,//匹配失败，终止状态
        Matched,//匹配陈宫

        //S2状态，传入基于S2状态可以进行的一系列状态转移
        S2(new Transition("fail", Matched), new Transition("success", Terminal)),

        //S1状态
        S1(new Transition("fail", S2), new Transition("success", Terminal)),

        //初始状态
        initial(new Transition("fail", S1), new Transition("success", Terminal));

        private Transition[] transitions;//当前状态的转移规则

        State(Transition... transitions) {
            this.transitions = transitions;
        }

        //状态转移方法
        public State transition(String eventType) {
            for (Transition transition : transitions) {
                if (transition.getEventType().equals(eventType)) {
                    return transition.getTargetState();
                }
            }
            //回到初始状态
            return initial;
        }
    }

    //定义一个状态转移类，包含当前引起状态转移的事件类型，以及转移的目标状态
    public static class Transition {

        private String eventType;

        private State targetState;

        public Transition(String eventType, State targetState) {
            this.eventType = eventType;
            this.targetState = targetState;
        }

        public String getEventType() {
            return eventType;
        }

        public State getTargetState() {
            return targetState;
        }
    }
}
